package yatan.cluster;

import org.apache.log4j.Logger;

import org.jgroups.Address;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.SuspectedException;
import org.jgroups.TimeoutException;
import org.jgroups.blocks.GroupRequest;
import org.jgroups.blocks.MessageDispatcher;
import org.jgroups.blocks.RequestHandler;
import org.jgroups.blocks.RequestOptions;
import org.jgroups.util.Rsp;

import yatan.common.utility.ParamUtility;
import yatan.common.utility.LogUtility;

/**
 * <p>
 * This class is the super class of all cluster nodes. It provides some common operation like start, stop, etc.
 * </p>
 * <p>
 * The cluster nodes use a third party component JGroup to communicate with each other inside a network. It actually
 * encapsulate the communication and protocol layer of this component. For detail information, please refer to <a
 * href="http://www.jgroups.org/">http://www.jgroups.org/</a>.
 * </p>
 * <p>
 * Also notice that this class use it self as a "Receiver" of JGroup, so subclasses could override any method from
 * {@link ReceiverAdapter} to process the messages received from other nodes.
 * </p>
 * <p>
 * <b> Thread Safety: </b> it's mutable and not thread safe. Cluster node is not supposed to be used in a multi-threaded
 * environment. It's more like a singleton.
 * </p>
 * @author EveningSun
 * @version 1.0
 */
public abstract class BaseNode implements RequestHandler {
    /**
     * <p>
     * The default value for clusterName field.
     * </p>
     */
    public static final String DEFAULT_CLUSTER_NAME = "YatanCralwerCluster";
    /**
     * <p>
     * The logger.
     * </p>
     */
    private Logger logger = Logger.getLogger(this.getClass());
    /**
     * <p>
     * The name of the cluster that the node will be joining in.
     * </p>
     */
    private String clusterName = DEFAULT_CLUSTER_NAME;
    private String jgroupConfig;
    /**
     * <p>
     * The JChannel object used to communicate.
     * </p>
     */
    private JChannel channel;
    private MessageDispatcher messageDispatcher;

    private Thread workThread;

    /**
     * <p>
     * Default empty constructor.
     * </p>
     */
    public BaseNode() {
    }

    /**
     * <p>
     * Start the node and connect to the cluster with the name {@link #clusterName}.
     * </p>
     * @throws NodeException if any error occurs while creating channel and connect to the cluster
     */
    public void start() throws NodeException {
        if (this.channel != null) {
            throw LogUtility.logError(this.logger, new IllegalStateException("The node is already started."));
        }

        this.logger.info("Try to start node " + this + "...");
        this.logger.info("This is a" + (this instanceof MasterNode ? " master " : " servant ") + "node");
        try {
            this.logger.debug("Try to create JChannel...");
            // create JChannel object
            this.channel = new JChannel(this.jgroupConfig);
            // create message dispatcher
            this.messageDispatcher = new MessageDispatcher(this.channel, null, null, this);

            // connect to the cluster
            this.logger.debug("Try to connect to the cluster: " + this.clusterName + "...");
            this.channel.connect(this.clusterName);
        } catch (ChannelException e) {
            throw LogUtility
                    .logError(this.logger, new NodeException("Cannot create channel or connect to the cluster '"
                            + this.clusterName + "': " + e.getMessage(), e));
        }

        this.logger.info("Node started succesfully! Address is " + this.channel.getAddressAsString()
                + ". Starting working thread...");

        // start work
        this.workThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    work();
                } catch (RuntimeException e) {
                    logger.fatal("Runtime exception occurred in work thread. Servant node will be stopped." + e, e);
                    stop();
                } catch (NodeException e) {
                    logger.fatal("Fatal exception occurred in work thread. Servant node will be stopped." + e, e);
                    stop();
                }
            }
        });
        this.workThread.setDaemon(true);
        this.workThread.setName(this.channel.getAddressAsString() + "#Working");
        this.workThread.start();
    }

    public void stop() {
        this.logger.info("Try to stop node " + this + "...");

        // FIXME: should stop work thread first here

        if (this.channel != null) {
            this.channel.disconnect();
            this.channel.close();
            this.channel = null;
        }

        this.logger.info("Node has been stopped.");
    }

    public boolean isStarted() {
        return this.channel != null;
    }

    public String getClusterName() {
        return clusterName;
    }

    public void setClusterName(String clusterName) {
        ParamUtility.checkStringNullEmpty(clusterName, "clusterName");

        this.clusterName = clusterName;
    }

    public void setJgroupConfig(String jgroupConfig) {
        this.jgroupConfig = jgroupConfig;
    }

    public String getJgroupConfig() {
        return jgroupConfig;
    }

    protected synchronized <T> T sendMessage(Address dest, NodeMessage nodeMessage, Class<T> expectedResultType, long timeout)
            throws TimeoutException, SuspectedException, NodeException {
        ParamUtility.checkNull(nodeMessage, "nodeMessage");
        if (!isStarted()) {
            throw new IllegalStateException("The node is not started yet.");
        }

        Message message = new Message(dest, this.channel.getAddress(), nodeMessage);

        Object result = null;
        if (dest == null) {
            getLogger().debug("Sending node message " + nodeMessage + " to all members...");
            for (Rsp response : this.messageDispatcher.castMessage(null, message,
                    new RequestOptions(GroupRequest.GET_MAJORITY, timeout)).values()) {
                if (response.getValue() != null) {
                    result = response.getValue();
                }
            }
            getLogger().debug("Response for node message " + nodeMessage + " all members receieved.");
        } else {
            getLogger().debug("Sending node message " + nodeMessage + " to " + message.getDest() + "...");
            result = this.messageDispatcher.sendMessage(message, new RequestOptions(GroupRequest.GET_FIRST, timeout));
            getLogger().debug("Response for node message " + nodeMessage + " to " + message.getDest() + " receieved.");
        }

        try {
            return expectedResultType.cast(result);
        } catch (ClassCastException e) {
            throw new NodeException("The response of message " + message + " should be type " + expectedResultType
                    + ", not " + result, e);
        }
    }

    protected JChannel getChannel() {
        return this.channel;
    }

    protected Logger getLogger() {
        return this.logger;
    }

    protected abstract void work() throws NodeException;
}
